package org.bikelab.se3.mr;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.bikelab.se3.rdf.reader.RdfReader;
import org.bikelab.se3.rdf.reader.RdfReaderException;
import org.bikelab.se3.rdf.reader.RdfReaderFactory;
import org.semanticweb.yars.nx.Node;

public class NQuadHbaseLoader {
	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value, Context context)
				throws IOException {
			try {
				RdfReader quadReader = RdfReaderFactory.getRdfReader(context.getConfiguration());
				quadReader.read(value.toString());
				while(quadReader.hasNext()) {
					Node[] quad = quadReader.next();
					String triple = quad[0].toN3() + " " + quad[1].toN3() + " " + quad[2].toN3() + " .";
					context.write(new Text(quad[3].toN3()), new Text(triple));
				}
			} catch (RdfReaderException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static class Reduce extends Reducer<Text, Text, NullWritable, Writable> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException {
			StringBuilder sb = new StringBuilder();
			Iterator<Text> list = values.iterator();
			while(list.hasNext()) {
				sb.append(list.next().toString());
				sb.append("\n");
			}
			Put put = new Put(Bytes.toBytes(key.toString()));
			put.add(Bytes.toBytes("content"), Bytes.toBytes("ntriple"), Bytes.toBytes(sb.toString()));
			try {
				context.write(NullWritable.get(), put);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
